package com.clsz.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * kafkamq演示
 */
@Component
public class MQConsumer {

    /**
     * 监听器回调方法
     * @param record 接收到的新消息
     * @param ack 消息回执
     */
    @KafkaListener(topics = "test02",groupId = "group-01")
    public void test02(ConsumerRecord<String, String> record, Acknowledgment ack) throws InterruptedException {

        System.out.println("====>consumer====>new message===>key="+record.key()+",value="+record.value());
        //模拟耗时任务,消费者执行
        System.out.println("消费者执行==============>模拟耗时任务");
        for (int i = 0; i < 10; i++) {
            System.out.println("===>任务执行度："+(i+1)+"/10");
            Thread.currentThread().sleep(1000);
        }
        System.out.println("end");
        //提交回执
        ack.acknowledge();
    }

}
